package com.atguigu.gmall.app.dws;

import com.atguigu.gmall.app.BaseSqlApp;
import com.atguigu.gmall.bean.KeywordStats;
import com.atguigu.gmall.common.Constant;
import com.atguigu.gmall.function.KWProductUdtf;
import com.atguigu.gmall.function.KWUdtf;
import com.atguigu.gmall.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/7 9:11
 */
public class DwsKeyWordProductStatsApp extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsKeyWordProductStatsApp().init(4004, 1, "DwsKeyWordProductStatsApp");
    }
    
    @Override
    protected void run(StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table product_stats(" +
                            "   stt string, " +
                            "   edt string, " +
                            "   sku_name string," +
                            "   click_ct bigint, " +
                            "   cart_ct bigint, " +
                            "   order_ct bigint " +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'properties.bootstrap.servers' = 'hadoop162:9092,hadoop163:9092, hadoop164:9092', " +
                            "   'properties.group.id' = 'DwsKeyWordProductStatsApp', " +
                            "   'topic' = '" + Constant.TOPIC_DWS_PRODUCT_STATS + "', " +
                            "   'scan.startup.mode' = 'latest-offset', " +
                            "   'format' = 'json' " +
                            ")");
        
        // 1. 过滤出来有效数据
        Table t1 = tEnv.sqlQuery("select * " +
                                     "from product_stats " +
                                     "where click_ct > 0 " +
                                     "or cart_ct > 0 " +
                                     "or order_ct > 0");
        tEnv.createTemporaryView("t1", t1);
        
        tEnv.createTemporaryFunction("ik_analyzer", KWUdtf.class);
        // 2. 对关键词进行分词
        Table t2 = tEnv.sqlQuery("select " +
                                     "  stt, edt, word, " +
                                     "  sum(click_ct) click_ct, " +
                                     "  sum(cart_ct) cart_ct, " +
                                     "  sum(order_ct) order_ct " +
                                     "from (select" +
                                        " stt, " +
                                        " edt, " +
                                        " word, " +
                                        " click_ct, " +
                                        " cart_ct,   " +
                                        " order_ct " +
                                        "from t1 " +
                                        ", lateral table(ik_analyzer(sku_name))) t" +
                                     "  group by stt, edt, word");
        tEnv.createTemporaryView("t2", t2);
        
        // 3. 把3个指标分别定位到3行
        tEnv.createTemporaryFunction("kw_product", KWProductUdtf.class);
    
        Table resultTable = tEnv.sqlQuery("select" +
                                        " stt, " +
                                        " edt," +
                                        " word keyword," +
                                        " source, " +
                                        " ct, " +
                                        " unix_timestamp() * 1000 ts " +
                                        "from t2 " +
                                        "join lateral table(kw_product(click_ct, cart_ct, order_ct)) on true");
        tEnv
            .toRetractStream(resultTable, KeywordStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink(Constant.CLICKHOUSE_DB, Constant.TABLE_KEYWORD_STATS, KeywordStats.class));
    
    
    }
}
